Repository navigation
[Mocha] Fix RabbitMQ recovery, publishing and outbox dispatch - #10543
Open
tobias-tengler wants to merge 32 commits into
Open
tobias-tengler wants to merge 32 commits into
tobias-tengler wants to merge 32 commits into
Conversation
Contributor
Patch coverage90.4% of changed lines covered (2147/2376) +35 more changed files; see the JSON below for uncovered lines. Uncovered changed lines (JSON){
"sha": "b0de3bc8eeafc9171ab9fbb54a1f17885c7d895d",
"files": [
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/ChannelOpenBlockedException.cs", "ranges": [[8, 8]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/RabbitMQQueueDescriptor.cs", "ranges": [[159, 159]] },
{ "path": "src/Mocha/src/Mocha/Consumers/Implementations/ReplyConsumer.cs", "ranges": [[69, 69]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/ThrowHelper.cs", "ranges": [[9, 9], [15, 15], [18, 20], [38, 38], [50, 50], [53, 53], [56, 58]] },
{ "path": "src/Mocha/src/Mocha/Middlewares/Consume/Retry/ConsumerRetryMiddleware.cs", "ranges": [[46, 49]] },
{ "path": "src/Mocha/src/Mocha.Transport.AzureServiceBus/AzureServiceBusReceiveEndpoint.cs", "ranges": [[300, 303]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConsumerManager.cs", "ranges": [[140, 141], [168, 168], [200, 200], [203, 207], [209, 209], [218, 218], [220, 220], [224, 224], [257, 257], [259, 259], [261, 261]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQDispatcher.Publishing.cs", "ranges": [[47, 48], [70, 71], [170, 178], [193, 203], [213, 222]] },
{ "path": "src/Mocha/src/Mocha/Middlewares/Receive/Redelivery/ReceiveRedeliveryMiddleware.cs", "ranges": [[53, 56]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQDispatcher.cs", "ranges": [[86, 86], [110, 111], [149, 153], [177, 180], [182, 183], [185, 185], [197, 197], [199, 201], [203, 203], [225, 226], [301, 303], [378, 378]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConnectionManager.cs", "ranges": [[115, 115], [137, 140], [143, 144], [160, 160], [162, 163], [175, 176], [200, 201], [215, 217], [241, 243], [286, 286], [316, 316], [360, 361], [370, 370], [398, 400], [473, 474], [496, 497], [515, 515], [517, 518], [520, 522], [524, 524], [565, 565], [568, 568], [588, 589], [650, 651], [653, 653], [674, 674], [686, 687], [689, 689], [750, 751], [758, 760]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQReceiveEndpoint.cs", "ranges": [[70, 71]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConsumer.cs", "ranges": [[102, 102], [269, 280], [348, 350], [460, 463], [478, 491], [563, 563], [565, 566], [682, 682], [684, 684], [702, 704], [750, 760]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQQueue.cs", "ranges": [[122, 123]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs", "ranges": [[80, 80], [158, 163], [166, 166]] },
{ "path": "src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs", "ranges": [[130, 130], [132, 133], [138, 140]] },
{ "path": "src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQAcknowledgementMiddleware.cs", "ranges": [[35, 35]] }
]
}Project coverage: 59.7% (324482/543839 lines) |
Contributor
|
✅ Nitro CLI E2E snapshots: no flow recordings changed vs |
tobias-tengler
force-pushed
the
tte/rabbitmq-consumer-reattach
branch
from
October 19, 2026 07:36
33654db to
aee642f
Compare
tobias-tengler
force-pushed
the
tte/rabbitmq-consumer-reattach
branch
from
October 27, 2026 07:48
60e1ed8 to
420a0ea
Compare
tobias-tengler
force-pushed
the
tte/rabbitmq-consumer-reattach
branch
2 times, most recently
from
November 4, 2026 08:23
0ec47a6 to
c7833e6
Compare
tobias-tengler
changed the base branch from
main
to
tte/mocha-rabbitmq-recreate-missing-queues
November 4, 2026 08:23
tobias-tengler
force-pushed
the
tte/rabbitmq-consumer-reattach
branch
3 times, most recently
from
November 5, 2026 15:16
2d19eb0 to
ed13ac3
Compare
tobias-tengler
marked this pull request as ready for review
November 7, 2026 16:36
Contributor
There was a problem hiding this comment.
🔵 Needs a closer look
Recovery, shared-channel confirmations, and shutdown draining require RabbitMQ/Postgres integration validation that cannot be performed in this dependency-free snapshot.
No open issues 🎉
What changed in this PR
This PR strengthens Mocha’s RabbitMQ delivery and recovery behavior and adds configurable concurrent dispatch to the Postgres outbox.
Changes:
- Restores consumers after connection or channel loss, preserves concurrency limits, and keeps reply queues available during recovery and graceful shutdown.
- Adds confirmed publishing, connection-loss retries, missing-queue detection, and startup timeout configuration.
- Adds concurrent outbox workers, expands recovery and lifecycle tests, and documents delivery guarantees and operational limits.
| File | Description |
|---|---|
website/content/docs/mocha/transports/rabbitmq.md |
Documents recovery, confirmations, shutdown, and queue lifetimes. |
website/content/docs/mocha/reliability.md |
Documents outbox concurrency and connection-pool considerations. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Topology/RabbitMQReplyQueueTests.cs |
Tests reply queue expiry and defaults. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Topology/RabbitMQBusDefaultsTests.cs |
Tests defaults for non-durable queues. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Routing/RabbitMQReceiveTopologyConventionTests.cs |
Updates generated reply queue expectations. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQTransportTests.cs |
Tests timeout propagation to both connections. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchProvisioningTests.cs |
Removes obsolete mocked dispatch tests. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Middlewares/RabbitMQAcknowledgementMiddlewareTests.cs |
Tests acknowledgement races and handler failures. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Helpers/TestBus.cs |
Bounds shutdown draining and cleans up failed startup. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Descriptors/RabbitMQDescriptorTests.cs |
Tests connection timeout validation. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Connection/RabbitMQConsumerManagerTests.cs |
Tests disposal after channel-close failures. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Connection/RabbitMQConsumerCapacityTests.cs |
Tests concurrency budgets across channel losses. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Connection/RabbitMQConsumerAttachmentIntegrationTests.cs |
Tests independent attachment, dependencies, and cancellation. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Connection/RabbitMQConnectionManagerBaseTests.cs |
Tests setup synchronization, timeouts, and recovery callbacks. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/UnifiedQueueBehaviorTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/TransportMiddlewareTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/TopologyRecoveryTests.cs |
Tests topology restoration after publisher reconnection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/TemporaryEndpointCleanupTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ShutdownTests.cs |
Tests draining, cancellation, requeueing, and reply availability. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/SendTests.cs |
Moves tests to the messaging collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/SchedulingTests.cs |
Moves tests to the messaging collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/RoutingKeyTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/RequestReplyTests.cs |
Updates reply lifetime and reconnection scenarios. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ReplyRecoveryTests.cs |
Tests request cancellation and replies during disconnection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ReplyQueueCleanupTests.cs |
Tests ownership-aware reply queue cleanup. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/PublishSubscribeTests.cs |
Moves tests to the messaging collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/PublishingReliabilityTests.cs |
Tests broker rejections and publishing failure isolation. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/PublishingConcurrencyTests.cs |
Tests concurrent sends and cancellation recovery. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/PublisherRecoveryTests.cs |
Tests publishing through blocking, disconnection, and restart. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/MissingQueueRecoveryTests.cs |
Updates queue and binding recovery scenarios. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/InboxTests.cs |
Moves tests to the messaging collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/FaultHandlingTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ExternalDestinationTests.cs |
Tests foreign destinations and missing-queue errors. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ExplicitTopologyTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ExclusiveQueueTests.cs |
Tests exclusive queue ownership and recovery. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ErrorQueueTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/EndpointMiddlewareTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/CustomHeaderTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConsumerLifecycleTests.cs |
Tests endpoint restart and uncooperative handlers. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConnectionStartupTests.cs |
Tests startup retries, failures, and connection names. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConnectionRecoveryTests.cs |
Verifies recovery without test-side publish retries. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConcurrencyTests.cs |
Updates recovery synchronization for concurrency tests. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ConcurrencyLimiterTests.cs |
Moves tests to the messaging collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/BusDefaultsIntegrationTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/BrokerRecoveryTests.cs |
Tests broker outages, credential changes, and resource cleanup. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/BatchingTests.cs |
Moves tests to the messaging collection. |
src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/AutoProvisionIntegrationTests.cs |
Moves tests to the topology collection. |
src/Mocha/test/Mocha.Tests/Transport/MessagingTransportLifecycleTests.cs |
Tests shutdown ordering and cleanup after failures. |
src/Mocha/test/Mocha.Tests/Threading/ExponentialBackoffTests.cs |
Tests retry delay bounds and randomization. |
src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxConcurrencyTests.cs |
Tests concurrent dispatch, idle workers, and claim release. |
src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/OutboxServiceRegistrationTests.cs |
Tests rejection of invalid concurrency settings. |
src/Mocha/src/Mocha/Transport/MessagingTransport.cs |
Drains application endpoints concurrently before replies. |
src/Mocha/src/Mocha/Runtime/MessagingRuntime.cs |
Cancels remaining shutdown work during disposal. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQTopologyProvisioner.cs |
Keeps exclusive queue provisioning on consumer connections. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQRoutingStrategy.cs |
Gives generated reply queues expiry-based lifetimes. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQReceiveTopology.cs |
Removes the receive topology wrapper. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQQueue.cs |
Adds expiry without auto-delete. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQMessagingTopology.cs |
Supports bypassing bus-level queue defaults. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/Configurations/RabbitMQQueueConfiguration.cs |
Adds internal control over default application. |
src/Mocha/src/Mocha.Transport.RabbitMQ/ThrowHelper.cs |
Centralizes setup, timeout, and returned-message exceptions. |
src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQReceiveEndpoint.cs |
Adds health reporting, reply dependencies, and guarded cleanup. |
src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessagingTransport.cs |
Applies startup timeouts and role-based connection names. |
src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs |
Retries connection losses and handles mandatory queue returns. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Mocha.Transport.RabbitMQ.csproj |
Updates RabbitMQ.Client to 7.2.2. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQAcknowledgementMiddleware.cs |
Handles channel closure during acknowledgement or rejection. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/RabbitMQQueueDescriptor.cs |
Qualifies the shared middleware configuration exception helper. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/RabbitMQMessagingTransportDescriptor.cs |
Validates and configures startup connection timeouts. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/IRabbitMQMessagingTransportDescriptor.cs |
Exposes the connection timeout API. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Connection/RabbitMQConsumerManager.ChannelConsumer.cs |
Queues deliveries and tracks handler cancellation across loss. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Configurations/RabbitMQTransportConfiguration.cs |
Stores startup timeout and supports named connections. |
src/Mocha/src/Mocha.Transport.RabbitMQ/Configurations/RabbitMQDefaultQueueOptions.cs |
Skips incompatible defaults for non-durable queues. |
src/Mocha/src/Mocha.Threading/ExponentialBackoff.cs |
Adds reusable randomized retry delay calculation. |
src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/PostgresOutboxOptions.cs |
Exposes maximum concurrent outbox dispatches. |
src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/PostgresMessageOutboxOptions.cs |
Carries concurrency into processor configuration. |
src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxServiceCollectionExtensions.cs |
Validates and propagates outbox concurrency. |
src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs |
Runs independent dispatch workers with coordinated cancellation. |
Review effort: Balanced
…hem away The transport now owns connection recovery instead of the client's automatic recovery, and re-attaches a consumer with backoff when the broker closes its channel or cancels it while the connection stays up. - Re-attach consumers whose channel was closed or whose subscription was cancelled, and declare the receive topology of every consumer before any consumer resumes on a connection. - Dispose lost connections and reconnect in the background with backoff. - Wait for the broker at startup for up to `ConnectTimeout`, two minutes by default, and fail at once on refused credentials, access or topology. - Publish with confirms, return messages sent to a missing queue, retry publishes on connection loss, and cap the number of rented publish channels. - Ack the redelivery of a message whose handler already completed without running the handler again. - Let reply queues expire instead of being deleted with their last consumer, and keep bus queue defaults away from them. - Declare exclusive queues only on the consumer connection. - Name connections by role and require RabbitMQ.Client 7.2.2.
Enforce connection timeouts, cancel consumer attachment during shutdown, and retry losses that arrive as recovery completes. Reuse consumer workers across reconnects, cancel handlers on channel loss, and expose blocked recovery health. Remove the completed-delivery cache and cover recovery, capacity limits, redelivery, and shutdown with regression tests.
Preserve compatible queue defaults and explicit argument overrides while skipping quorum and stream defaults for non-durable queues. Cover both queue type configuration forms with regression tests. Clarify that shutdown cancellation ends handler draining while channel cleanup continues independently.
Share publisher channels per endpoint and bound outstanding publish attempts while waiting for individual broker confirmations. Add configurable Postgres outbox concurrency with independent workers so blocked dispatches do not stall idle capacity. Preserve cancellation behavior with explicit loop termination conditions. Add concurrency and cancellation coverage and document the publishing and outbox settings.
Wait for consumer capacity before reattaching and for the local reply consumer before dispatching requests. Cancel pending connection waiters and drain connection setup during disposal, with idempotent consumer shutdown and forced-stop escalation. Keep request deadlines active until dispatch completes, including when a response or fault arrives before dispatch finishes. Apply explicit quorum replication defaults to generated reply queues so requests can recover through surviving cluster nodes. Respect broker defaults for unspecified queue types and delivery limits. Honor explicitly configured limits with RabbitMQ 3 compatibility, and preserve existing queues and their messages during provisioning. Cover broker process crashes, repeated restarts, quorum leader failover, connection failover, cancellation, disposal races, request deadlines, and finite or unlimited redelivery with regression tests. Validation: 223 targeted tests passed across RabbitMQ 3 and 4 and core request deadlines. Documentation formatting and rendered pages checked.
Add correlated recovery, retry, fault, publish-wait, and drain diagnostics. Preserve connection and channel identity across recovery, clarify health failures, and correct conversation tracing. Remove log-capture assertions and retain behavioral coverage for delivery, recovery, cancellation, and faults. Document scope export configuration.
Configure the cluster fixture to return disconnected-consumer deliveries after one second and verify the broker is running after each restart. Add coverage for an unacknowledged event held by a killed consumer node.
tobias-tengler
force-pushed
the
tte/rabbitmq-consumer-reattach
branch
from
November 10, 2026 15:14
60eae49 to
d55287e
Compare
RabbitMQ transport - Replace the connection manager base class with one internal RabbitMQConnectionManager: one connect routine with a deadline per attempt, a ready signal, and a recovery that only the first loss report starts. A connection attempt against an endpoint that never answers now ends. - Publish on an exclusive channel lease from a pool per connection, with at most 100 channels. A channel is reused only after the broker gave a verdict on it, so a rejected or cancelled publish no longer affects other publishes. - Add the PublishTimeout transport setting (default 2 minutes). A send or publish that the broker does not confirm in time fails with the new TransportUnavailableException. - Keep the connection during a broker memory or disk alarm and do not send waiting messages again. - Run each consumer as one lifecycle loop. Cancelling or stopping one consumer no longer aborts the shared connection, and a consumer whose handlers block reports an error only after a grace period. - Remove the wait of receive endpoints for the reply endpoint. - Classify failures as transient or permanent. A message the broker can never accept is no longer retried without end. - Rename the log events and log one event per fact. - RabbitMQDispatcher, RabbitMQConsumerManager and RabbitMQConnectionManagerBase are no longer public. Outbox - Dispatch with independent workers. A failing dispatch no longer cancels the others, which sent an already sent message again. - Add DispatchTimeout, MaxAttempts and MaxRetryDelay. An unavailable transport does not use up attempts, and an exhausted message is logged as Critical with its id, type and destination. Core and other transports - Stopping with a cancelled token still cleans up in the Postgres and Azure Service Bus transports. - A failed transport start stops what already started. - The message bus health check takes a name, failure status, tags and a timeout, and uses the ready tag by default. - The transport circuit breaker logs under its own category. - ExponentialBackoff applies jitter at the cap as well. Tests and docs - Recovery tests prove the loss they recover from, with shared broker fault helpers and one timeout policy. Mock-based tests are replaced by tests against a broker. - Document recovery, the publish timeout, broker alarms, the outbox options and the health check.
tobias-tengler
changed the base branch from
tte/mocha-rabbitmq-recreate-missing-queues
to
main
November 11, 2026 07:42
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
RabbitMQ consumers could stop receiving after a connection or channel loss, outbound sends could complete without broker confirmation, and the PostgreSQL outbox could lose or duplicate messages around a broker outage.
This PR reworks how the RabbitMQ transport connects, publishes and consumes, and how the outbox reacts to a failed dispatch.
RabbitMQ transport
Connections
ConnectTimeoutsetting and fails immediately on rejected credentials, access or topology.Publishing
PublishTimeout(default 2 minutes, infinite allowed). A publish the broker does not confirm in time fails with the newTransportUnavailableException.Consuming
Logging
PostgreSQL outbox
MaxConcurrentMessages. The default remains 1. Higher values use additional database connections and allow completion out of order. A failing dispatch does not affect the other workers.DispatchTimeout(default 30 seconds) and a host stop release the row without counting an attempt.MaxRetryDelay(default 10 minutes). AfterMaxAttempts(default 10) the row stays in the table and a Critical entry is logged.Core, hosting and other transports
AddMessageBustakesname,failureStatus,tagsandtimeout, with the tagreadyand a timeout of 10 seconds by default.Mocha.TransportCircuitBreakerMiddleware.ExponentialBackoff.CalculateDelayapplies jitter at the cap as well.Breaking changes
RabbitMQDispatcher,RabbitMQConsumerManagerandRabbitMQConnectionManagerBaseare internal now, as are theDispatcherandConsumerManagerproperties ofRabbitMQMessagingTransport.IRabbitMQConnectionProviderstays public.TransportUnavailableExceptionafterPublishTimeout.AddMessageBus(IHealthChecksBuilder, ...)has new optional parameters, which is binary breaking. The check now carries the tagready.Known limitations
basic.nackis not retried. The outbox counts it as an attempt.ScheduledMessageWorker.StopAsyncdoes not wait for a dispatch the transport already accepted, so a scheduled message can be sent again after a restart.